package com.flitsneak.rocket.producer;

import com.flitsneak.common.entity.Info;
import com.flitsneak.common.entity.Messages;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.Date;

/**
 * @author flitsneak
 * @email flitsneak@gmail.com
 * @date 2022/8/10 14:47
 * @description
 */
@Slf4j
@RequiredArgsConstructor
@RestController
@RequestMapping("/rocket")
public class RocketProducer {
    private final RocketMQTemplate rocketMQTemplate;

    /**
     * 同步异步消息 单向广播 延迟消息
     * @param num 数量
     */
    @GetMapping("/sendMessage/{num}")
    public String sendMessage(@PathVariable Integer num) {
        for (int i = 0; i < num; i++) {
            Messages messages = new Messages();
            messages.setId(i);
            messages.setTime(new Date());
            messages.setDes("hello rocket!");

            //rocketMQTemplate.convertAndSend("topic1",messages);

            //同步消息
            //rocketMQTemplate.syncSend("topic1",messages);

            //异步消息
//            rocketMQTemplate.asyncSend("topic1", messages, new SendCallback() {
//                @Override
//                public void onSuccess(SendResult sendResult) {
//                    log.info("success!");
//                }
//
//                @Override
//                public void onException(Throwable throwable) {
//                    log.error("fail");
//                }
//            },1000);

            //单向发送
            //rocketMQTemplate.sendOneWay("topic1",messages);

            //延时消息
            //rocketMQTemplate.syncSend("topic1", MessageBuilder.withPayload(messages).build(),2000,2);

            //批量消息
//            ArrayList<messages> list = new ArrayList<>();
//            rocketMQTemplate.syncSend("topic1",list,1000);

            //带tag发送
            rocketMQTemplate.syncSend("topic1:tag1", messages);

            log.info("rocket发送消息：{}", messages);
        }

        return "ok!";
    }

    /**
     * 顺序消息
     */
    @GetMapping("/sendOrderMessage")
    public String sendOrderMessage() {
        ArrayList<Info> infos = new ArrayList<>(9);
        Info info = new Info();
        info.setId(1);
        info.setDes("1开始");
        infos.add(info);

        info = new Info();
        info.setId(2);
        info.setDes("2开始");
        infos.add(info);

        info = new Info();
        info.setId(3);
        info.setDes("3开始");
        infos.add(info);

        info = new Info();
        info.setId(3);
        info.setDes("3结束");
        infos.add(info);

        info = new Info();
        info.setId(2);
        info.setDes("2结束");
        infos.add(info);

        info = new Info();
        info.setId(1);
        info.setDes("1结束");
        infos.add(info);

        rocketMQTemplate.setMessageQueueSelector((list, message, o) -> {
            int num = Integer.parseInt(o.toString());
            int i = num % list.size();
            return list.get(i);
        });

        infos.forEach(c -> {
            rocketMQTemplate.syncSendOrderly("topic1", c, c.getId().toString());
            log.info("发送顺序消息：{}", c);
        });

        return "OKK!";
    }

    /**
     * 事务消息
     * @return
     */
    @GetMapping("/sendTransactional")
    public String sendTransactional() {
        Info info = new Info();
        info.setTime(new Date());
        info.setDes("hello");
        info.setId(99);
        rocketMQTemplate.sendMessageInTransaction("topic1", MessageBuilder.withPayload(info).build(),null);

        return "OKK!!";
    }
}
